package com.jbm.core.common;

import com.jbm.core.common.ResponseRecvCallback.ResponseRecvCallback;
import com.jbm.core.iface.IAcross;
import com.jbm.core.pb.PBIAcross;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * Created by root on 17-5-31.
 */
public class MessageQueue
{
	public static final MessageQueue g_mq = new MessageQueue();

	private BlockingQueue m_task;
	public ResponseRecvCallback m_rrcallback;

	public MessageQueue()
	{
	}

	public void Init(int task, int thread)
	{
		m_task = new ArrayBlockingQueue(task);

		m_rrcallback = new ResponseRecvCallback()
		{
			public void PushResponse(long start, long cur, PBIAcross.IAcross request, IAcross content, PBIAcross.IAcross response, Object exinfo, boolean isfirst)
			{
				if ( response == null )
					return;

				try
				{
					if (response.getCode() != 200)
					{
						m_task.put(exinfo);

						if ( isfirst )
						{
							//database insert
						}
					}else
						Drop((PBIAcross.IAcross)exinfo);
				}catch (Exception e)
				{
				}
			}
		};

		for ( int i = 0; i < thread; i ++ )
		{
			MQThread t = new MQThread();
			t.start();
		}
	}

	private class MQThread extends Thread
	{
		@Override
		public void run()
		{
			while (true)
			{
				try
				{
					PBIAcross.IAcross req = (PBIAcross.IAcross)m_task.take();

					AutoCall.g_autocall.Call(req,m_rrcallback, false, 0, null, MainServer.Server_type_Server);

					sleep(5);
				}catch (Exception e)
				{
				}
			}
		}
	}

	private void Drop(PBIAcross.IAcross req)
	{
		//database delete
	}

	public void Add(PBIAcross.IAcross req)
	{
		try
		{
			m_task.put(req);
		}catch (Exception e)
		{
		}
	}
}
